package org.dbstar.scheduler.beans;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

import org.apache.commons.lang.StringUtils;
import org.dbstar.scheduler.control.JobController;
import org.dbstar.scheduler.control.exception.DaoException;
import org.dbstar.scheduler.identifier.DefaultIdentifier;
import org.dbstar.scheduler.identifier.Identifier;
import org.dbstar.scheduler.job.JobStatus;
import org.dbstar.scheduler.utils.JobUtils;


/**
 * 基于复合类型的Job的执行结果
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-5-7
 */
public class CompoundJobResult extends JobResult {
	private static final long serialVersionUID = -3668055470283031365L;

	private List<SubJobResult> results = new ArrayList<CompoundJobResult.SubJobResult>();

	public List<SubJobResult> getResults() {
		return Collections.unmodifiableList(results);
	}

	@Override
	public synchronized void init(Job job, JobController jobController) throws DaoException {
		for (JobDefine jd : ((CompoundJob) job).getSubJobs()) {
			if (!jd.isSkip()) {
				Job j = jobController.getJob(jd.getIdentifier());
				if (!updateSubJobResult(j)) {
					addSubJobResult(j, jd.getDependencies());
				}
			}
		}
	}

	public boolean updateSubJobResult(Job subJob) {
		return updateSubJobResult(subJob.getIdentifier(), subJob.getStatus(), subJob.isSucceed());
	}

	public synchronized boolean updateSubJobResult(Identifier identifier, JobStatus status, boolean succeed) {
		for (SubJobResult r : results) {
			if (r.identifier.equals(identifier)) {
				r.status = status;
				r.succeed = succeed;
				return true;
			}
		}
		return false;
	}

	public void addSubJobResult(Job subJob, Set<Integer> dependencies) {
		addSubJobResult(subJob.getIdentifier(), subJob.getStatus(), subJob.isSucceed(), dependencies);
	}

	public synchronized void addSubJobResult(Identifier identifier, JobStatus status, boolean succeed,
			Set<Integer> dependencies) {
		results.add(new SubJobResult(identifier, status, succeed, dependencies));
	}

	@Override
	public synchronized boolean isSucceed() {
		if (results.isEmpty()) {
			// 此时尚未初始化
			return false;
		}
		// 只有所有子Job都执行成功，复合Job才算成功
		for (SubJobResult r : results) {
			if (!JobUtils.isFinished(r.status) || !r.succeed) {
				return false;
			}
		}
		return true;
	}

	/**
	 * 检查是否所有子任务都运行结束，结束状态包括：kill,cancel,error和finish
	 * 
	 * @return
	 */
	public synchronized boolean isFinished() {
		for (SubJobResult r : results) {
			if (!JobUtils.isFinished(r.status)) {
				return false;
			}
		}
		return true;
	}

	/**
	 * 获得依赖条件满足的子Job
	 * 
	 * @return
	 */
	public synchronized Set<Identifier> getReadyJobs() {
		Set<Identifier> set = new HashSet<Identifier>();
		for (SubJobResult r : results) {
			if (r.status == JobStatus.init && checkDependency(r)) {
				set.add(r.identifier);
			}
		}
		return set;
	}

	public synchronized Set<Identifier> getInitJobs() {
		Set<Identifier> set = new HashSet<Identifier>();
		for (SubJobResult r : results) {
			if (r.status == JobStatus.init) {
				set.add(r.identifier);
			}
		}
		return set;
	}

	private boolean checkDependency(SubJobResult r) {
		if (r.dependencies != null) {
			for (int index : r.dependencies) {
				if (!results.get(index - 1).succeed) {
					return false;
				}
			}
		}
		return true;
	}

	@Override
	public void reExecute(Job job, boolean failedOnly, JobController jobController) throws DaoException {
		super.reExecute(job, failedOnly, jobController);
		for (JobDefine jd : ((CompoundJob) job).getSubJobs()) {
			if (jd.isSkip()) {
				if (!failedOnly) {
					if (!updateSubJobResult(jd.getIdentifier(), JobStatus.init, false)) {
						addSubJobResult(jd.getIdentifier(), JobStatus.init, false, jd.getDependencies());
					}
				}
			} else {
				Job j = jobController.getJob(jd.getIdentifier());
				if (j.reExecute(failedOnly, jobController)) {
					jobController.updateJob(j);
					jobController.changeJobStatus(j.getIdentifier(), JobStatus.init);
					j = jobController.getJob(jd.getIdentifier());
				}
				if (!updateSubJobResult(j)) {
					addSubJobResult(j, jd.getDependencies());
				}
			}
		}
	}

	@Override
	protected void write(JSONObject object) {
		super.write(object);
		if (!results.isEmpty()) {
			JSONArray array = new JSONArray();
			for (SubJobResult r : results) {
				JSONObject obj = new JSONObject();
				obj.element("identifier", r.identifier.getStringIdentifier());
				obj.element("status", r.status.name());
				obj.element("succeed", r.succeed);
				if (r.dependencies != null && !r.dependencies.isEmpty()) {
					obj.element("dependencies", StringUtils.join(r.dependencies, ','));
				}
				array.add(obj);
			}
			object.element("results", array);
		}
	}

	@Override
	protected void read(JSONObject object) {
		super.read(object);
		results.clear();
		JSONArray array = object.optJSONArray("results");
		if (array != null) {
			for (int i = 0, size = array.size(); i < size; i++) {
				JSONObject obj = array.getJSONObject(i);
				Identifier identifier = DefaultIdentifier.valueOf(obj.getString("identifier"));
				JobStatus status = JobStatus.valueOf(obj.getString("status"));
				boolean succeed = obj.getBoolean("succeed");
				Set<Integer> dependencies = null;
				String ds = obj.optString("dependencies", null);
				if (ds != null) {
					dependencies = new HashSet<Integer>();
					for (String s : StringUtils.split(ds, ',')) {
						dependencies.add(Integer.valueOf(s));
					}
				}
				results.add(new SubJobResult(identifier, status, succeed, dependencies));
			}
		}
	}

	@Override
	public JobResult clone() throws CloneNotSupportedException {
		CompoundJobResult result = (CompoundJobResult) super.clone();
		if (results != null) {
			result.results = new ArrayList<CompoundJobResult.SubJobResult>(results.size());
			for (SubJobResult r : results) {
				result.results.add(r.clone());
			}
		}
		return result;
	}

	public static class SubJobResult implements Serializable, Cloneable {
		private static final long serialVersionUID = 8073976165824294873L;

		private Identifier identifier;
		private JobStatus status;
		private boolean succeed;
		private Set<Integer> dependencies;

		public SubJobResult(Identifier identifier, JobStatus status, boolean succeed, Set<Integer> dependencies) {
			super();
			this.identifier = identifier;
			this.status = status;
			this.succeed = succeed;
			this.dependencies = dependencies;
		}

		public Identifier getIdentifier() {
			return identifier;
		}

		public JobStatus getStatus() {
			return status;
		}

		public boolean isSucceed() {
			return succeed;
		}

		public Set<Integer> getDependencies() {
			return dependencies;
		}

		@Override
		public SubJobResult clone() throws CloneNotSupportedException {
			SubJobResult result = (SubJobResult) super.clone();
			if (dependencies != null) {
				result.dependencies = new HashSet<Integer>(dependencies);
			}
			return result;
		}
	}
}
